package com.bigdata.spark.mallapp_realtime


import java.lang
import java.text.SimpleDateFormat
import java.util.Date
import com.bigdata.spark.util.OffsetManagerUtil
import com.alibaba.fastjson.{JSON, JSONObject}
import com.bigdata.spark.util.{MyKafkaUtil, MyRedisUtil, OffsetManagerUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer


object mall_log_Exactly_once {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("DauApp")
    val ssc: StreamingContext = new StreamingContext(conf,Seconds(5))
    var topic:String = "gmall_event_0523"

    var groupId = "atguigu"

    //从Redis中获取Kafka分区偏移量。Redis中的key是topic+groupId，值是map类型的TopicPartition+offset
    val offsetMap: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic,groupId)


    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsetMap!=null && offsetMap.size >0){
      //如果Redis中存在当前消费者组对该主题的偏移量信息，那么从执行的偏移量位置开始消费
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,offsetMap,groupId)
    }else{
      //如果Redis中没有当前消费者组对该主题的偏移量信息，那么还是按照配置，从最新位置开始消费
      recordDStream = MyKafkaUtil.getKafkaStream(topic,ssc,groupId)
    }

    //获取当前采集周期从Kafka中消费的数据的起始偏移量以及结束偏移量值,目的是，逻辑完毕之后更新偏移量。获取结束偏移量，要从
    //现在读取到的recordDStream这个离散化流获取。这个用getKafkaStream创建的离散化流底层是kafkaRDD，这才可以取到起始结束偏移量值。

    //创建读取数据用的array
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  //就是new Array[offsetRange](0)


    //可以实现只取值，不提交改变的算子：transform。
    val offsetDStream: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {

        //取到起始结束偏移量
        //因为recodeDStream底层封装的是KafkaRDD，混入了HasOffsetRanges特质，这个特质中提供了可以获取偏移量范围的方法
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  //这个属性是private的，不能随便访问
        //RDD里没有offsetRanges方法。要访问它的父类来。所以要把RDD转换成HasOffsetRanges。RDD强制类型转换为父类HasOffsetRanges。
        //tips：存在继承关系才能强制转换。kafkaRDD子，HasOffsetRanges是父。（所以叫做KafkaRDD混入了HasOffsetRanges特质）
        rdd
      }
    }
    //offsetRanges这样就取得了，一次获取的是所有分区内所有partition的分区，主题，起始结束

    //   其中单条offsetRange: 数据，获得了读取kafka时的分区，主题，起始结束。
    //    val topic: String,
    //    val partition: Int,
    //    val fromOffset: Long,
    //    val untilOffset: Long) extends Serializable


    val StringlogDS: DStream[String] = offsetDStream.map(
      ConsumerRecord => ConsumerRecord.value()

    )


    //保存的时候需要幂等性。

    StringlogDS.foreachRDD(  //foreachRDD的用法：里面的参数是个迭代器。
      // 这是保存数据用的算子。对本时间间隔，执行一次。这里是5秒。在一个为批次的处理时间间隔里， DStream只产生一个RDD
      rdd=>{


        rdd.foreach(
          record=>{
            println(record)
          }

        )

            println("第一层")
            println(rdd,rdd.getClass)
            //提交偏移量到Redis中，以分区为单位。
            //println("提交偏移量到Redis中，以分区为单位。")
            OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)

      }


    )



//    offsetDStream.foreachRDD{  //这是保存数据用的算子。对本时间间隔，执行一次。这里是5秒。在一个为批次的处理时间间隔里， DStream只产生一个RDD
//      rdd=>{
//        //以分区为单位对数据进行处理，就是一个批保存一次。
//        rdd.foreachPartition{
//          resit=>{
//
//
//            resit.map{
//              ConsumerRecord=>{
//
//                println(ConsumerRecord.value(),"777777777777777777")
//                //ConsumerRecord.value()
//              }
//
//            }
//            println(resit,resit.getClass,"99999999999999999999999999999")
//          }
//            //将数据批量的保存
//        }
//
//        //提交偏移量到Redis中，以分区为单位。
//        //println("提交偏移量到Redis中，以分区为单位。")
//        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)
//      }
//    }




    ssc.start()
    ssc.awaitTermination()


  }
}
